package com.behelpful.demo.search.config.binlog;

import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.EventType;
import com.github.shyiko.mysql.binlog.event.TableMapEventData;
import com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer;
import com.behelpful.demo.search.event.ListenerAllocate;
import com.behelpful.demo.search.event.deserializer.DeleteRowsDeserializer;
import com.behelpful.demo.search.event.deserializer.UpdateRowsDeserializer;
import com.behelpful.demo.search.event.deserializer.WriteRowsDeserializer;
import lombok.AllArgsConstructor;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.core.annotation.Order;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.lang.reflect.Field;
import java.util.Map;

/**
 * 监听MySQL binlog
 * CommandLineRunner SpringBoot启动后执行的代码（后置初始化）
 *
 * @author Alay
 * @date 2021-08-05 00:13
 * @project behelpful
 */
@Order
@Component
@AllArgsConstructor
@EnableConfigurationProperties(value = MySqlConnection.class)
public class BinLogRunner implements CommandLineRunner {
    /**
     * 监听器（这里只注册了一个监听器,然后在其中进行逻辑分发时间处理）
     */
    private final ListenerAllocate listenerAllocate;
    private final MySqlConnection mySQLConnection;

    @Async
    @Order
    @Override
    public void run(String... args) throws Exception {
        EventDeserializer eventDeserializer = new EventDeserializer();

        // 由于下面的自定义系列化需要反系列化对象的 tableMapEventByTableId 字段值,而此字段是私有的,所以通过反射拿
        Field field = eventDeserializer.getClass().getDeclaredField("tableMapEventByTableId");
        field.setAccessible(true);
        Map<Long, TableMapEventData> tableMapEventByTableId = (Map<Long, TableMapEventData>) field.get(eventDeserializer);

        // 自定义反系列化类 (读写更新)
        eventDeserializer.setEventDataDeserializer(EventType.EXT_WRITE_ROWS, new WriteRowsDeserializer(tableMapEventByTableId));
        eventDeserializer.setEventDataDeserializer(EventType.EXT_UPDATE_ROWS, new UpdateRowsDeserializer(tableMapEventByTableId));
        eventDeserializer.setEventDataDeserializer(EventType.EXT_DELETE_ROWS, new DeleteRowsDeserializer(tableMapEventByTableId));

        BinaryLogClient logClient = new BinaryLogClient(mySQLConnection.getHost(), mySQLConnection.getPort(),
                mySQLConnection.getUserName(), mySQLConnection.getPassword());
        logClient.setServerId(1);
        logClient.setEventDeserializer(eventDeserializer);
        // 监听器
        logClient.registerEventListener(listenerAllocate);
        /**
         * 此处可能出现异常处理,AuthenticationException 因为 MySQL 访问密码加密协议的不同问题
         * 关于此问题作者的交流讨论： https://github.com/shyiko/mysql-binlog-connector-java/issues/240
         */
        logClient.connect();
    }
}